#!/usr/bin/env python
# -*- coding:utf-8 -*-
import json
import sys
import subprocess
import os
import re
import psutil
import time
import requests
from copy import deepcopy
from datetime import datetime, timedelta
from pybloom_live import ScalableBloomFilter
from secret import batch_encrypt_decrypt_dict
from aes import contact_aes, name_aes
from apscheduler.schedulers.blocking import BlockingScheduler
from datastore.mongodb import MongoDBClientWrapper

reload(sys)
sys.setdefaultencoding('utf-8')


'''
====漏斗：===

bidding表导出数据 多少条
有多少条联系方式数：
有多少条联系方式数据且有nameId：
根据nameId和contact去重有多少

手机、固话、邮箱、其他 分别统计上面3个维度


入联系方式表的有多少 source、contact、nameId
属于更新的有多少、新增的有多少、绝对新增有多少

上线产品端有多少


'''


file_name = 'bidding历史漏斗'
not_in_contact_f = open('not_in_contact_f','w')
def serializable_bloom(result_dic):
    for k, v in result_dic.items():
        if isinstance(v, ScalableBloomFilter):
            result_dic[k] = result_dic[k].count
        if isinstance(v, dict):
            serializable_bloom(v)
    return result_dic

class Utils(object):
    '''工具类，用于封装'''

    @staticmethod
    def decrypt_one_record(record_dic, just_strf_time=False):
        """
        just_strf_time ：True只需要格式化时间而不用解密其他的
        递归解密一条记录
        """
        new_result = dict()

        for key, value in record_dic.items():
            if type(value) == datetime:
                value = value.strftime("%Y-%m-%d %H:%M:%S")
            new_result[key] = value
        if not just_strf_time:
            try:
                new_result = batch_encrypt_decrypt_dict([new_result], 'decrypt')[0]
            except:
                pass
        return new_result

    @staticmethod
    def bloom_filter_fac(initial_capacity=1000000, error_rate=0.001, mode=2):
        '''统一控制bloom配置
        @param initial_capacity: int
        @param error_rate: float
        @param mode: int  2慢但是节省内存、4快耗内存
        @return: bloom
        '''
        return ScalableBloomFilter(initial_capacity=initial_capacity, error_rate=error_rate, mode=mode)  # 慢点但是省内存

    @staticmethod
    def get_date_range(ndays=0, start_time=None, end_time=None):
        '''
        @param ndays: int 时间间隔
        @param start_time: 开始时间字符串 str '2020-12-10'
        @param end_time: 结束时间
        @return: tuple of datetime
        '''
        if start_time and end_time:
            start_time = datetime.strptime(start_time, "%Y-%m-%d")
            end_time = datetime.strptime(end_time, "%Y-%m-%d")
        else:
            end_time = datetime.now()
            # today = datetime(year=2021, month=6, day=4)
            n_day = timedelta(days=ndays)
            start_time = end_time - n_day
        _min = datetime(year=start_time.year, month=start_time.month, day=start_time.day)
        _max = datetime(year=end_time.year, month=end_time.month, day=end_time.day)
        return _min, _max

    @staticmethod
    def generate_abs_path(file_name, current_py_path):
        '''
        @param file_name: 文件名 str
        @return: 绝对路径文件名  str
        '''
        file_dir = "/mnt5/dsf/"
        file_name = file_dir + file_name

        current_time_str = datetime.now().strftime('%Y-%m-%d_%H')  # '2021-05-28_15'
        abs_file_name = '{}_{}_{}'.format(current_py_path.rsplit('.', 1)[0], current_time_str,file_name)
        return abs_file_name

    @staticmethod
    def run_shell(shell_cmd):
        '''
        运行shell命令
        @param shell_cmd:  "sort -u {} | wc -l".format(file_name)  str
        @return: shell命令的返回值 str
        '''
        p = subprocess.Popen(shell_cmd, shell=True, stdout=subprocess.PIPE, stderr=subprocess.STDOUT)
        result_str = p.stdout.read().decode('utf-8')
        return result_str

abs_file_name = Utils.generate_abs_path(file_name,current_py_path=__file__)

class DingtalkRobot(object):
    '''
    将统计结果发送给钉钉，日志写入文件（待做）
        # 电商群机器人
    web_hook_url = 'https://oapi.dingtalk.com/robot/send?access_token=db25f78eac3de543f3f40ffc42cec8736ea6320408c84c65aaac46853a97f632'
    # 联系方式测试机器人
    web_hook_url = 'https://oapi.dingtalk.com/robot/send?access_token=1c23bf41ce0e5a7b453129f719c710e395eeb50445a190212b2eb04c109d5490'
    # 联系方式群正式机器人
    web_hook_url = 'https://oapi.dingtalk.com/robot/send?access_token=6a732fad118e7d46b1662b2c9a623591e1441e637de8ae9fb528d30ce449f2ba'

    '''

    def __init__(self):
        self.ding_robot_file = open(abs_file_name+'_robot','w')


    def send_msg_to_dd(self, web_hook_url, title, mk_text):
        '''
        发送mk_text到钉钉
        @param web_hook_url: 机器人链接  str
        @param title: 主题名  str
        @param mk_text: 钉钉内容  str
        @return: None
        '''

        msg_body = {
            "msgtype": "markdown",
            "markdown": {
                "title": title,
                "text": mk_text
            },
        }
        headers = {
            "Content-Type": "application/json;charset=utf-8"
        }
        retry_count = 0
        while retry_count < 10:
            try:
                response = requests.post(url=web_hook_url, data=json.dumps(msg_body), headers=headers)
                if response.status_code == 200:
                    self.ding_robot_file.write("{} 发送钉钉信息成功：{}\n".format(datetime.now(), response.status_code))

                    break
                else:
                    self.ding_robot_file.write("{}  返回状态码不正常：{}\n".format(datetime.now(), response.status_code))
            except Exception as e:
                self.ding_robot_file.write("{}  请求异常：{}\n".format(datetime.now(), e))
                retry_count += 1
                time.sleep(2)

    def main(self,web_hook_url,title,mk_text):
        self.send_msg_to_dd(web_hook_url, title, mk_text)
        self.ding_robot_file.close()


class ExportAndTjBase(object):
    '''统计、导出任务实现 的基类'''

    def __init__(self, file_name):
        self.db = MongoDBClientWrapper()
        # 脚本开始时间
        self.start_time = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
        self.file_name = file_name
        self.init_file(file_name)
        self.log('***  the start', suffix=' ***')

    def init_file(self, file_name, mod='w'):
        '''
        初始化需要写入的文件
        @param file_name: 绝对路径文件名  str
        @param mod: 文件读写模式  str
        @return: None
        '''
        dir_path = file_name.rsplit('/', 1)[0]
        if not os.path.exists(dir_path):
            os.makedirs(dir_path)
        self.file_dir_path = dir_path

        self.write_log_f = open('{}_log'.format(file_name), mod)
        self.write_dc_f = open('{}_dc'.format(file_name), mod)
        self.write_tj_f = open('{}_tj'.format(file_name), mod)

    def close_file(self):
        '''
        关闭所有文件流
        @return: None
        '''
        self.log('\n***  the end ', suffix='***')
        self.write_log_f.close()
        self.write_dc_f.close()
        self.write_tj_f.close()

    def memory_check(self, max_percent=95):
        '''
        内存检测
        @param max_percent: int
        @return: None
        '''
        memory_info = psutil.virtual_memory()
        memory_percent = memory_info.percent
        if memory_percent > max_percent:  # 内存占比
            self.log()
            time.sleep(60)
            if memory_percent > 98:
                exit()
        return memory_percent

    def log(self, log_info='', suffix=''):
        '''
        @param log_info: 要记录的日志信息
        @param suffix: 单行日志尾部信息
        @return: None
        '''
        memory_percent = self.memory_check()
        current_time = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
        log_info += '\tmemory_percent：{}\tnow：{} {}\n'.format(memory_percent, current_time, suffix)
        self.write_log_f.write(log_info)
        # 让开发者立即看到脚本效果
        self.write_log_f.flush()

    def handle_index_log(self, index, save_result_dic):
        '''
        根据索引打日志
        @param index: 当前下标
        @param save_result_dic: 当前统计结果
        @return: None
        '''
        if index == 0:
            self.log('find is ok!')
        if index % 10000 == 0:
            if index % 1000000 == 0:
                self.save_tj_info(save_result_dic)
            self.log('index：{}'.format(index))

    def save_tj_info(self, result_dic):
        '''
        保持统计信息到统计文件
        @param result_dic: 统计信息 dict
        @return: None
        '''
        result_dic = deepcopy(result_dic)
        current_time = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
        self.write_tj_f.write('save at {}\n{}\n\n\n'.format(current_time, json.dumps(serializable_bloom(result_dic))))
        self.write_tj_f.flush()

    def get_records_crusor(self, config):
        '''
        获得查询结果cursor对象
        @param config: config = {
            'col_name': 集合名称 str
            'filter': 筛选条件 dic
            'projection': 投影 list of str
        @return: cursor对象
        '''
        col = self.db.p_col(config['col_name'])
        filter = config['filter']
        projection = config.get('projection')
        params = {
            'filter':filter
        }
        if projection is not None:
            params.update(projection=projection)
        # 0 mobile 手机 ； 1 phone 固话 ；2 email
        if config.get('limit'):
            records_crusor = col.find(**params).limit(config['limit']).sort([("$natural", -1)])
        else:
            records_crusor = col.find(**params) # .limit(10000).sort([("$natural",-1)]),
        return records_crusor

    def get_data_from_mongo(self, config, save_result_dic=None):
        '''
        从mongo中查出数据
        @param config: config = {
            'col_name': 集合名称 str
            'filter': 筛选条件 dic
            'projection': 投影 list of str
        @param save_result_dic: 存放统计数据的dict
        @return:result_dic
        '''
        # 0 mobile 手机 ； 1 phone 固话 ；2 email
        records_crusor = self.get_records_crusor(config)
        for index, record_dic in enumerate(records_crusor):
            save_result_dic = self.handle_one_mongo_data(record_dic, save_result_dic)
            self.handle_index_log(index, save_result_dic)
        else:
            self.log('get_data_from_mongo over ')
            self.save_tj_info(save_result_dic)

        return save_result_dic

    def handle_one_mongo_data(self, record_dic, save_result_dic=None):
        '''
        对一条查询结果进行处理
        导出符合条件的，或者进行统计
        @param record_dic: 数据库一条记录  dict
        @param save_result_dic: 存放统计结果  dict
        @return: save_result_dic  dict
        '''
        pass

    def run(self):
        '''
        主逻辑
        @return:None
        '''
        pass

    def main(self):
        '''
        运行入口
        @return:None
        '''
        self.run()
        self.close_file()


class ExportAndTj(ExportAndTjBase):
    def __init__(self, *args, **kwargs):
        super(ExportAndTj, self).__init__(*args, **kwargs)
        # todo other col
        self.contact_col = self.db.p_col('contact')


    def run(self):
        file_name = '/mnt5/dsf/bidding/bidding_history_bak'
        save_result_dic = {
            'bidding_count':0, #bidding表导出数据 多少条
            'bidding_contact_count':0, # 多少条联系方式数：
            'bidding_contact_and_nameId_count': 0,  # 多少条联系方式数据且有nameId
            'bidding_contact_and_nameId_dup': Utils.bloom_filter_fac(),  # 根据nameId和contact去重有多少

            'bidding_mobile_count':0, # 多少条手机数：
            'bidding_mobile_and_nameId_count':0, # 多少条手机数且有nameId
            'bidding_mobile_and_nameId_dup':Utils.bloom_filter_fac(), #手机根据nameId和contact去重有多少
            'bidding_phone_count':0, # 多少条固话数：
            'bidding_phone_and_nameId_count': 0,  # 多少条固话数且有nameId
            'bidding_phone_and_nameId_dup': Utils.bloom_filter_fac(),  # 固话根据nameId和contact去重有多少
            'bidding_email_count':0, # 多少条邮箱数：
            'bidding_email_and_nameId_count': 0,  # 多少条邮箱数且有nameId
            'bidding_email_and_nameId_dup': Utils.bloom_filter_fac(),  # 邮箱根据nameId和contact去重有多少
            'bidding_other_count':0, # 多少条其他联系方式数：
            'bidding_other_and_nameId_count': 0,  #多少条其他联系方式数且有nameId
            'bidding_other_and_nameId_dup': Utils.bloom_filter_fac(),  # 其他根据nameId和contact去重有多少

            'into_contact_count':0, #入联系方式表的有多少 source、contact、nameId
            'into_phone_count':0, #入联系方式表的有多少 source、contact、nameId
            'into_mobile_count':0, #入联系方式表的有多少 source、contact、nameId
            'into_email_count':0, #入联系方式表的有多少 source、contact、nameId
            'into_other_count':0, #入联系方式表的有多少 source、contact、nameId
            'contact_update_count':0, # 属于更新的有多少、新增的有多少、绝对新增有多少
            'contact_add_count':0,
            'contact_abs_add_count':0,
        }
        index = 0
        with open(file_name, 'r') as f, open('tj_result', 'w') as log_f:
            for line in f:
                line = line.strip()
                if line:
                    index += 1
                    print(line)
                    save_result_dic['bidding_count'] += 1
                    dic = json.loads(line)
                    contact_ls = dic.get('biddingProjectContact')
                    biddingSource = dic.get('biddingSource','')
                    if contact_ls:
                        for tmp_contact_dic in contact_ls:
                            save_result_dic['bidding_contact_count'] += 1
                            contact = tmp_contact_dic.get('contact', '')
                            contactType = tmp_contact_dic.get('contactType', '')
                            nameId = tmp_contact_dic.get('nameId', '')
                            _id = "{}_{}_{}".format(contact, contactType,biddingSource,nameId)
                            if contactType == 0:
                                save_result_dic['bidding_mobile_count'] += 1
                            elif contactType == 1:
                                save_result_dic['bidding_phone_count'] += 1
                            elif contactType == 2:
                                save_result_dic['bidding_email_count'] += 1
                            else:
                                save_result_dic['bidding_other_count'] += 1

                            if nameId:
                                save_result_dic['bidding_contact_and_nameId_count'] += 1
                                is_in = save_result_dic['bidding_contact_and_nameId_dup'].add(_id)
                                contact_flag = False
                                if not is_in: #如果不重复
                                    filter = {
                                        "nameId":nameId,
                                        "contact":contact_aes.encrypt(contact),
                                    }
                                    if biddingSource:
                                        filter.update({'contactChannelList.contactSource':biddingSource})
                                    projection = [ "contactType", "import_update_time","create_time","contactChannelList"]

                                    record_dic = self.contact_col.find_one(filter=filter,projection=projection)
                                    if record_dic:
                                        save_result_dic['into_contact_count'] += 1
                                        if contactType == 0:
                                            save_result_dic['into_phone_count'] += 1
                                            save_result_dic['bidding_mobile_and_nameId_count'] += 1
                                            save_result_dic['bidding_mobile_and_nameId_dup'].add(_id)
                                        elif contactType == 1:
                                            save_result_dic['into_mobile_count'] += 1
                                            save_result_dic['bidding_phone_and_nameId_count'] += 1
                                            save_result_dic['bidding_phone_and_nameId_dup'].add(_id)
                                        elif contactType == 2:
                                            save_result_dic['into_email_count'] += 1
                                            save_result_dic['bidding_email_and_nameId_count'] += 1
                                            save_result_dic['bidding_email_and_nameId_dup'].add(_id)
                                        else:
                                            save_result_dic['into_other_count'] += 1
                                            save_result_dic['bidding_other_and_nameId_count'] += 1
                                            save_result_dic['bidding_other_and_nameId_dup'].add(_id)
                                        create_t = datetime(2021,8,1)
                                        if record_dic.get('create_time') and record_dic.get('create_time')>= create_t:
                                            save_result_dic['contact_add_count'] += 1
                                            contactChannelList = record_dic.get('contactChannelList', [])
                                            abs_flag = len(contactChannelList) == 1
                                            if abs_flag:
                                                save_result_dic['contact_abs_add_count'] += 1
                                        elif record_dic.get('import_update_time') and record_dic.get('import_update_time')>= create_t:
                                            save_result_dic['contact_update_count'] += 1
                                    else:
                                        not_in_contact_f.write('{}\t{}\n'.format(nameId,contact))
                                        not_in_contact_f.flush()
                                        if contactType == 0:
                                            save_result_dic['bidding_mobile_and_nameId_count'] += 1
                                            save_result_dic['bidding_mobile_and_nameId_dup'].add(_id)
                                        elif contactType == 1:
                                            save_result_dic['bidding_phone_and_nameId_count'] += 1
                                            save_result_dic['bidding_phone_and_nameId_dup'].add(_id)
                                        elif contactType == 2:
                                            save_result_dic['bidding_email_and_nameId_count'] += 1
                                            save_result_dic['bidding_email_and_nameId_dup'].add(_id)
                                        else:
                                            save_result_dic['bidding_other_and_nameId_count'] += 1
                                            save_result_dic['bidding_other_and_nameId_dup'].add(_id)
                    if index % 100000 == 0:
                        self.save_tj_info(deepcopy(save_result_dic))
                        log_f.write(json.dumps(serializable_bloom(deepcopy(save_result_dic)))+'\n\n\n')
            else:
                self.save_tj_info(deepcopy(save_result_dic))
                log_f.write(json.dumps(serializable_bloom(deepcopy(save_result_dic))) + '\n\n\n')




def run(abs_file_name):
    export_and_tj = ExportAndTj(abs_file_name)
    export_and_tj.main()


def scheduler_run(abs_file_name):
    scheduler = BlockingScheduler()
    scheduler.add_job(func=run, trigger='cron',
                      args=(abs_file_name,),
                      hour='20', minute='52', id='dd_run')
    scheduler.start()


if __name__ == '__main__':

    run(abs_file_name)
    # scheduler_run(abs_file_name)
